package com.foreverwin.jzymes.archive;

import com.foreverwin.jzymes.archive.domain.DataArchive;
import com.foreverwin.jzymes.archive.service.IDataArchiveAdviceService;
import com.foreverwin.jzymes.framework.util.SpringUtils;
import com.github.pagehelper.Page;
import com.github.pagehelper.PageHelper;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;

public class ArchiveTask implements Runnable {
    private static int DEFAULT_PAGE_SIZE = 100;
    private static int MAX_PAGE_SIZE = 200;
    private static int DEFAULT_PAGES = 10;
    private Logger logger = LoggerFactory.getLogger( ArchiveTask.class );

    private DataArchive dataArchive;
    private Map<String,Object> paramPool;
    private IDataArchiveAdviceService dataArchiveAdviceService;


    public ArchiveTask(DataArchive dataArchive, Map<String,Object> paramPool ){
        this.dataArchive = dataArchive;
        this.paramPool = paramPool;
        this.dataArchiveAdviceService = SpringUtils.getBean(IDataArchiveAdviceService.class);
    }

    @Override
    public void run() {
        List<String> syncFields = buildSyncFields();
        List<String> allFields = buildAllFields();
        List<String> primaryKeys = buildPrimaryKeys();

        if( allFields.size() == 0 ){
            return;
        }

        if( syncFields.size() == 0 ){
            syncFields = allFields;
        }

        /**最底层子表同步未正常结束，下次再同步时需进行删除*/
        if( StringUtils.isNotBlank( dataArchive.getReferTable() ) ){
            int count = dataArchiveAdviceService.doDeleteArchive( paramPool, dataArchive.getArchiveTable(), dataArchive.getQueryCriteria() );
            updateArchiveCount( dataArchive.getTableName(), -count );
            logger.info( "表{}删除数据-{}", dataArchive.getTableName(), count );
        }

        List<DataArchive> referDataArchiveList = getReferDataArchiveList();
        if( "Y".equals( dataArchive.getDeleteAfterSync() ) ){
            doArchiveDelete( referDataArchiveList, syncFields, allFields, primaryKeys );
        }else{
            doArchiveSync( referDataArchiveList, syncFields, allFields, primaryKeys );
        }
    }

    public void doArchiveSync( List<DataArchive> referDataArchiveList, List<String> syncFields, List<String> allFields, List<String> primaryKeys ){

        int pageSize = comptePageSize();
        List<Map<String,Object>> archiveDataList = null;
        if( dataArchive.getReferTable() == null ){
            int offset = dataArchive.getCount().intValue();
            int pages = 0;
            int count = 1;
            do{
                archiveDataList = doQueryOffsetPage( offset, pageSize, allFields );
                pages = computePages( pages, archiveDataList );

                doArchive( referDataArchiveList, syncFields, primaryKeys, archiveDataList, false );

                offset = offset + pageSize;
                count++;
            }while( pages >= count );
        }else{
            int pageNum = 1;
            int pages = 0;
            do{
                archiveDataList = doQueryPage( pageNum, pageSize, allFields );
                pages = computePages( pages, archiveDataList );

                doArchive( referDataArchiveList, syncFields, primaryKeys, archiveDataList, false );

                pageNum++;
            }while( pages >= pageNum );
        }
    }

    public void doArchiveDelete( List<DataArchive> referDataArchiveList, List<String> syncFields, List<String> allFields, List<String> primaryKeys ){
        int pageNum = 1;
        int pageSize = comptePageSize();
        int pages = 0;
        List<Map<String,Object>> archiveDataList = null;
        do{
            archiveDataList = doQueryPage( pageNum, pageSize, allFields );
            pages = computePages( pages, archiveDataList );

            doArchive( referDataArchiveList, syncFields, primaryKeys, archiveDataList, true );

            pages --;
        }while( pages >= pageNum );
    }

    public void doArchive( List<DataArchive> referDataArchiveList, List<String> syncFields, List<String> primaryKeys, List<Map<String,Object>> archiveDataList, boolean delete ){
        List<Future> futureList = new ArrayList<>();
        logger.info( "表{}开始同步", dataArchive.getTableName() );
        if( referDataArchiveList.size() > 0 ){
            for( Map<String,Object> data : archiveDataList ){
                ArchiveDataTask archiveDataTask = new ArchiveDataTask( dataArchive, referDataArchiveList, syncFields, primaryKeys, data, delete );
                FutureTask future = new FutureTask( archiveDataTask, null );
                futureList.add( future );
                new Thread( future ).start();
            }
            for( Future<Integer> future : futureList ){
                try {
                    future.get();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }
            }
        }else{
            Long startTime = System.currentTimeMillis();
            dataArchiveAdviceService.doBatchDeleteArchive( archiveDataList, dataArchive.getArchiveTable(), primaryKeys );
            logger.info( "表{}批次删除耗时-{}", dataArchive.getTableName(), System.currentTimeMillis() - startTime );
            int count = dataArchiveAdviceService.doBatchInsertArchive( archiveDataList, dataArchive.getArchiveTable(), syncFields );
            updateArchiveCount( dataArchive.getTableName(), count );
            if( delete ){
                dataArchiveAdviceService.doBatchDelete( archiveDataList, dataArchive.getTableName(), primaryKeys );
            }
            logger.info( "表{}批次写入耗时-{}", dataArchive.getTableName(), System.currentTimeMillis() - startTime );
        }
        logger.info( "表{}同步完成-{}", dataArchive.getTableName(), archiveDataList.size() );
    }

    public int comptePageSize(){
        if( dataArchive.getPageSize() == 0 ){
            return DEFAULT_PAGE_SIZE;
        }else if( dataArchive.getPageSize() > MAX_PAGE_SIZE ){
            return MAX_PAGE_SIZE;
        }else{
            return dataArchive.getPageSize().intValue();
        }
    }

    public int computePages( int pages, List<Map<String,Object>> archiveDataList ){
        if( pages <= 0 ){
            if( archiveDataList.size() > 0 ){
                Page pageInfo = (Page) archiveDataList;
                pages = pageInfo.getPages();
                int pageNum = pageInfo.getPageNum();
                pages = pages - pageNum + 1;
            }
            if( dataArchive.getReferTable() == null && pages > 0 ){
                pages = dataArchive.getPages().intValue() > pages ? pages : dataArchive.getPages().intValue();
                pages = pages > 0 ? pages :  DEFAULT_PAGES;
            }
        }
        return pages;
    }

    public List<String> buildSyncFields(){
        if( dataArchive.getSyncField() != null && dataArchive.getSyncField() != "" ){
            return Arrays.asList( dataArchive.getSyncField().split(",") );
        }else{
            return new ArrayList<>();
        }
    }

    public List<String> buildAllFields(){
        return dataArchiveAdviceService.findTableFields( dataArchive.getTableName() );
    }

    public List<String> buildPrimaryKeys(){
        return Arrays.asList( dataArchive.getPrimaryKey().split(",") );
    }

    public String converFieldsString( List<String> fields ){
        return String.join( ",", fields );
    }

    public List<DataArchive> getReferDataArchiveList(){
        return dataArchiveAdviceService.findReferDataArchiveList( dataArchive.getTableName() );
    }

    public int updateArchiveCount( String table, int increment ){
        return dataArchiveAdviceService.updateArchiveCount( table, increment );
    }

    public List<Map<String,Object>> doQueryOffsetPage( int offset, int pageSize, List<String> syncFields ){
        PageHelper.offsetPage( offset, pageSize );
        if( StringUtils.isNotBlank( dataArchive.getOrderClause() ) ){
            PageHelper.orderBy( dataArchive.getOrderClause() );
        }
        List<Map<String,Object>> data = doQuery( syncFields );
        return data;
    }

    public List<Map<String,Object>> doQueryPage( int pageNum, int pageSize, List<String> syncFields ){
        if( StringUtils.isNotBlank( dataArchive.getOrderClause() ) ){
            PageHelper.startPage( pageNum, pageSize, dataArchive.getOrderClause() );
        }else{
            PageHelper.startPage( pageNum, pageSize );
        }
        List<Map<String,Object>> data = doQuery( syncFields );
        return data;
    }

    public List<Map<String,Object>> doQuery( List<String> fields ){
        Map<String,Object> paramMap = new HashMap<>( paramPool );
        paramMap.put( "table", dataArchive.getTableName() );
        paramMap.put( "fields", converFieldsString( fields ) );
        paramMap.put( "queryCriteria", dataArchive.getQueryCriteria() );

        return dataArchiveAdviceService.doQuery( paramMap );
    }

}
